package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Demo8_TimeWindowJoin {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        tenv.executeSql("CREATE TABLE t1                  (\n" +
                "  uid BIGINT                                 \n" +
                "  ,event_id STRING                           \n" +
                "  ,action_time BIGINT                        \n" +
                "  ,rt1 as to_timestamp_ltz(action_time,3)    \n" +
                "  ,watermark for rt1 as rt1                  \n" +
                ") WITH (                                    \n" +
                "  'connector' = 'kafka',                    \n" +
                "  'topic' = 'ss-1',                         \n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'doit44_g1',      \n" +
                "  'scan.startup.mode' = 'latest-offset',    \n" +
                "  'value.format' = 'json',                  \n" +
                "  'value.fields-include' = 'EXCEPT_KEY'     \n" +
                ")");

        tenv.executeSql("CREATE TABLE t2                  (  \n" +
                "  uid BIGINT                                \n" +
                "  ,face_status int                           \n" +
                "  ,ts bigint                                 \n" +
                "  ,rt2 as to_timestamp_ltz(ts,3)             \n" +
                "  ,watermark for rt2 as rt2                  \n" +
                ") WITH (                                    \n" +
                "  'connector' = 'kafka',                    \n" +
                "  'topic' = 'ss-2',                         \n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'doit44_g2',      \n" +
                "  'scan.startup.mode' = 'latest-offset',    \n" +
                "  'value.format' = 'json',                  \n" +
                "  'value.fields-include' = 'EXCEPT_KEY'     \n" +
                ")");

        // 每5秒的识别行为，和识别结果按照相同uid关联
        tenv.executeSql(
                        " SELECT                                                               "+
                        "                                                                      "+
                        " o1.*,                                                                "+
                        " o2.*                                                                 "+
                        "                                                                      "+
                        " FROM (                                                               "+
                        "                                                                      "+
                        "     SELECT                                                           "+
                        "       *                                                              "+
                        "     from table(                                                      "+
                        "         tumble(table t1, descriptor(rt1),interval '5' second)        "+
                        "     )                                                                "+
                        " ) o1                                                                 "+
                        "                                                                      "+
                        " JOIN                                                                 "+
                        "                                                                      "+
                        " (                                                                    "+
                        "                                                                      "+
                        "     SELECT                                                           "+
                        "       *                                                              "+
                        "     from table(                                                      "+
                        "         tumble(table t2, descriptor(rt2),interval '5' second)        "+
                        "     )                                                                "+
                        " ) o2                                                                 "+
                        "                                                                      "+
                        "                                                                      "+
                        " ON o1.window_start = o2.window_start                                 "+
                        " and o1.window_end=o2.window_end                                      "+
                        " and o1.uid = o2.uid                                                  "
        ).print();



    }

}
